package com.nepu.spark.realtime.app

import java.util

import com.alibaba.fastjson.{JSON, JSONObject}
import com.nepu.spark.realtime.constant.KafkaConstant
import com.nepu.spark.realtime.util.{KafkaOffsetUtils, MyKafkaUtils, MyRedisUtils}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.spark.SparkConf
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{HasOffsetRanges, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import redis.clients.jedis.Jedis

/**
  * 本类是做业务数据的分离
  *
  * @author chenshuaijun
  * @create 2022-10-31 10:37
  */

/**
  * 1、首先是创建上下文对象
  * 2、从Redis中读取偏移量
  * 3、从kafka中消费数据
  * 3、提取kafka偏移量
  * 4、转换数据结构
  * 5、分流数据
  *   5.1、事实表的数据不进行格式的转换直接发送到对应的tipic中
  *   5.2、维度数据发送到对应的topic中
  * 6、刷写kafka
  * 7、提交offset
  */
object OdsDBDataDiversion {

  def main(args: Array[String]): Unit = {

    // 1、创建上下文对象
    val sparkConf: SparkConf = new SparkConf().setMaster("local[4]").setAppName("ods_db_data_diversion")
    val ssc = new StreamingContext(sparkConf, Seconds(3))

    val topic: String = KafkaConstant.ODS_BASE_DB_TOPIC
    val groupId: String = "ods_base_db"

    // 2、从Redis中读取偏移量
    val offset: Map[TopicPartition, Long] = KafkaOffsetUtils.getOffset(topic, groupId)
    var kafkaDStream: InputDStream[ConsumerRecord[String, String]] = null
    if (offset != null && offset.nonEmpty) {
      // 如果Redis中维护的有offset就使用，Redis中的offset进行消费
      kafkaDStream = MyKafkaUtils.getKafkaDStream(ssc, topic, groupId, offset)
    } else {
      // 如果Redis中没有维护offset就使用kafka中维护的offset
      kafkaDStream = MyKafkaUtils.getKafkaDStream(ssc, topic, groupId)
    }
    // 3、提取offset
    var offsetRanges: Array[OffsetRange] = null
    val offsetRangesDStream: DStream[ConsumerRecord[String, String]] = kafkaDStream.transform(
      rdd => {
        offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
        rdd
      }
    )

    // 4、转换数据结构
    val jsonObjDStream: DStream[JSONObject] = offsetRangesDStream.map(
      consumerRecord => {
        val value: String = consumerRecord.value()
        val jSONObject: JSONObject = JSON.parseObject(value)
        jSONObject
      }
    )

    // 5、对业务数据进行分流
    jsonObjDStream.foreachRDD(
      rdd => {
        // 首先配置表的动态清单
        val jedis: Jedis = MyRedisUtils.getJedisClient()
        // 获取事实表的清单
        val factTableNameSet: util.Set[String] = jedis.smembers(KafkaConstant.FACT_TABLE_NAME)
        // 获取维度表的清单
        val dimTableNameSet: util.Set[String] = jedis.smembers(KafkaConstant.DIM_TABLE_NAME)
        // 关闭jedis连接
        jedis.close()
        // 将表做成广播变量
        val factTablesBroadcase: Broadcast[util.Set[String]] = ssc.sparkContext.broadcast(factTableNameSet)
        val dimTablesBroadcae: Broadcast[util.Set[String]] = ssc.sparkContext.broadcast(dimTableNameSet)

        rdd.foreachPartition(
          iter => {
            val jedis: Jedis = MyRedisUtils.getJedisClient()
            for (jsonObj <- iter) {
              // 先获取当前maxwell同步过来的操作的类型
              val operationType: String = jsonObj.getString("type")

              val operationValue: String = operationType match {
                case "insert" => "I" // 插入操作
                case "update" => "U" // 更新操作
                case "delete" => "D" // 删除操作
                case "bootstrap-insert" => "I" // 全量同步操作
                case _ => null // 其它操作
              }
              // 过滤掉我们不需要的操作
              if (operationValue != null) {
                // 因为数据是不需要拆分的，所以我们直接进行数据的分流
                // 先拿到表名
                val tableName: String = jsonObj.getString("table")
                // 事实表分流
                if (factTablesBroadcase.value.contains(tableName)) {
                  // 拿到对应的数据，因为我们将数据存储成字符串，所以直接将数据提取成String
                  val data: String = jsonObj.getString("data")
                  // 这里以表为基准，以操作为单位，讲数据分离到不同的topic中
                  val factTopic: String = s"DWD_${tableName.toUpperCase}_$operationValue"
                  MyKafkaUtils.send(factTopic, data)
                }

                // 维度表处理
                if (dimTablesBroadcae.value.contains(tableName)) {
                  // 首先拿到对应的数据，因为我们需要使用到id，所以需要先转换成Object类型
                  val dataObject: JSONObject = jsonObj.getJSONObject("data")
                  // 拿到对应的id
                  val id: String = dataObject.getString("id")
                  // 拼接Redis中的key
                  val redisKey: String = s"DIM:${tableName.toUpperCase}:$id"
                  jedis.set(redisKey, dataObject.toJSONString)
                }
              }
            }
            // 关闭Redis连接
            jedis.close()
            // 刷写kafka的缓冲区
            MyKafkaUtils.flush()
          }
        )
        // 手动提交offset
        KafkaOffsetUtils.saveOffset(topic, groupId, offsetRanges)
      }
    )


    ssc.start()
    ssc.awaitTermination()

  }

}
